package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

var(
	err error
	kafkaReceiver sarama.Consumer
)

func main(){
	kafkaReceiver,err =sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
	partitionList, err := kafkaReceiver.Partitions("createorder") // 根据topic取到所有的分区
	if err != nil {
		fmt.Printf("fail to get list of partition:err%v\n", err)
		return
	}

	if len(partitionList) == 0{
		fmt.Printf("没有设置分区信息")
		return
	}

		// 针对每个分区创建一个对应的分区消费者
		pc, err := kafkaReceiver.ConsumePartition("createorder", int32(partitionList[0]), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partitionList[0], err)
			return
		}
		defer pc.AsyncClose()

		for{
			msg :=<- pc.Messages()
			fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
		}
}